package com.atguigu.day09;

import com.atguigu.util.Event;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

// 超时订单的检测
public class Example2 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        var stream = env
                .addSource(new SourceFunction<Event>() {
                    @Override
                    public void run(SourceContext<Event> ctx) {
                        ctx.collectWithTimestamp(new Event("order-1", "pay-order", 4000L), 4000L);
                        ctx.collectWithTimestamp(new Event("order-2", "create-order", 2000L), 2000L);
                        ctx.collectWithTimestamp(new Event("order-1", "create-order", 1000L), 1000L);

                        ctx.emitWatermark(new Watermark(9000L));
                        ctx.collectWithTimestamp(new Event("order-2", "pay-order", 10000L), 10000L);

                        ctx.collectWithTimestamp(new Event("order-3", "create-order", 11000L), 11000L);
                        ctx.collectWithTimestamp(new Event("order-3", "pay-order", 12000L), 12000L);
                    }

                    @Override
                    public void cancel() {

                    }
                });

        stream
                .keyBy(r -> r.key)
                .process(new StateMachine())
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static class StateMachine extends KeyedProcessFunction<String, Event, String> {
        private static final Map<Tuple2<String, String>, String> stateMachine;

        static {
            stateMachine = new HashMap<>();

            stateMachine.put(Tuple2.of("初始状态", "create-order"), "订单未支付");
            stateMachine.put(Tuple2.of("初始状态", "pay-order"), "订单已支付");
            stateMachine.put(Tuple2.of("订单未支付", "pay-order"), "订单已支付");
            stateMachine.put(Tuple2.of("订单未支付", "timeout"), "订单超时");
        }

        private ValueState<String> currentState;

        @Override
        public void open(Configuration parameters) throws Exception {
            currentState = getRuntimeContext().getState(
                    new ValueStateDescriptor<>(
                            "current-state",
                            Types.STRING
                    )
            );
        }

        @Override
        public void processElement(Event in, Context ctx, Collector<String> out) throws Exception {
            if (currentState.value() == null) {
                currentState.update("初始状态");
            }

            var nextState = stateMachine.get(Tuple2.of(currentState.value(), in.type));

            if (nextState != null) {
                if (nextState.equals("订单已支付")) {
                    out.collect(in.key + "订单已支付");
                } else if (nextState.equals("订单未支付")) {
                    ctx.timerService().registerEventTimeTimer(in.ts + 5000L);
                }

                currentState.update(nextState);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            var nextState = stateMachine.get(Tuple2.of(currentState.value(), "timeout"));
            if (nextState != null && nextState.equals("订单超时")) {
                out.collect(ctx.getCurrentKey() + "超时未支付");
                currentState.update(nextState);
            }
        }
    }
}
